package tss.hbase;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.List;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import tss.datastore.meta.EntityDefinition;
import tss.datastore.meta.MetaPersister;

public class HBaseMetaPersister implements MetaPersister {

    private static String entityTableName = "entity";
    private static byte[] defaultFamily = Bytes.toBytes("cf");    
    private static byte[] contentColumn = Bytes.toBytes("ct");
    

    private HBaseDriver hBaseDriver;

    public HBaseMetaPersister(HBaseDriver hBaseDriver) {
        this.hBaseDriver = hBaseDriver;
        createMetaTable();
    }

    private void createMetaTable() {
        hBaseDriver.doWithAdminConnection(new HBaseAdminAction() {

            @Override
            public void perform(HBaseAdmin admin) throws IOException {
                if (!admin.tableExists(entityTableName)) {
                    HTableDescriptor hTableDescriptor = new HTableDescriptor(entityTableName);
                    hTableDescriptor.addFamily(new HColumnDescriptor(defaultFamily));
                    admin.createTable(hTableDescriptor);
                }
            }
        });
    }

    @Override
    public void writeEntity(final EntityDefinition newEntityDefinition) {
        hBaseDriver.doWithTable(entityTableName, new HBaseTableAction() {

            @Override
            public void perform(HTable hTable) throws IOException {
                Put put = new Put(Bytes.toBytes(newEntityDefinition.getId()));                
                put.add(defaultFamily, contentColumn, entityDefinitionToBytes(newEntityDefinition));
                hTable.put(put);
            }
        });
    }

    public void clearAll() {
        hBaseDriver.doWithAdminConnection(new HBaseAdminAction() {

            @Override
            public void perform(HBaseAdmin admin) throws IOException {
                if (admin.tableExists(entityTableName)) {
                    admin.disableTable(entityTableName);
                    admin.deleteTable(entityTableName);
                }
            }
        });
        createMetaTable();
    }

    @Override
    public EntityDefinition loadEntity(final long entityId) {

        EntityDefinition result = hBaseDriver.queryTable(entityTableName, new HBaseTableQuery<EntityDefinition>() {

            @Override
            public EntityDefinition perform(HTable hTable) throws IOException {
                Get get = new Get(Bytes.toBytes(entityId));
                get.addColumn(defaultFamily, contentColumn);
                
                Result result = hTable.get(get);                
                if (!result.isEmpty()) {
                    List<KeyValue> content = result.getColumn(defaultFamily, contentColumn);
                    return bytesToEntityDefinition(content.get(0).getValue());
                } else {
                    return null;
                }

            }
        });
        return result;

    }
    
    private EntityDefinition bytesToEntityDefinition(byte[] bytes) {
        try (ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes))){
            return (EntityDefinition) ois.readObject();
        } catch (IOException | ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }
    
    private byte[] entityDefinitionToBytes(EntityDefinition entityDefinition) {
        try (ByteArrayOutputStream bio = new ByteArrayOutputStream();
             ObjectOutputStream oos = new ObjectOutputStream(bio)) {
            oos.writeObject(entityDefinition);
            oos.flush();
            return bio.toByteArray();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void deleteEntity(final long entityId) {
        hBaseDriver.doWithTable(entityTableName, new HBaseTableAction() {
            
            @Override
            public void perform(HTable hTable) throws IOException {
                Delete delete = new Delete(Bytes.toBytes(entityId));
                hTable.delete(delete);                
            }
        });        
    }

}
